package cn.gwm.flink.streaming.function.sink;

import cn.gwm.flink.streaming.beans.trip.CarTripOutput;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.core.lang.Dict;
import cn.hutool.core.util.ObjectUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @author byh
 */
public class SinkHbase extends RichSinkFunction<List<List<CarTripOutput>>> {
    Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        org.apache.hadoop.conf.Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", ConfigLoader.get("zookeeper.quorum"));
        configuration.set("hbase.zookeeper.property.clientPort", ConfigLoader.get("zookeeper.clientPort"));
        configuration.setInt("hbase.rpc.timeout", 6000);
        configuration.setInt("hbase.client.operation.timeout", 6000);
        configuration.setInt("hbase.client.scanner.timeout.period", 6000);
        connection = ConnectionFactory.createConnection(configuration);
    }

    @Override
    public void close() throws Exception {
        super.close();
        connection.close();
    }

    @Override
    public void invoke(List<List<CarTripOutput>> list2, Context context) throws Exception {
        super.invoke(list2, context);
        for(List<CarTripOutput> list:list2){
            for (CarTripOutput carTripOutput : list) {
                carTripOutput = CarTripOutput.calculationAverageSpeed(carTripOutput);
                carTripOutput = CarTripOutput.convertDecimalFormat(carTripOutput);
                Table table = connection.getTable(TableName.valueOf(carTripOutput.getTriptype()));
                Put put = new Put(Bytes.toBytes(carTripOutput.getDeviceId() + carTripOutput.getItem_time()));
                Dict dict = Dict.parse(carTripOutput);
                CarTripOutput finalCarTripOutput = carTripOutput;
                dict.keySet().forEach(fieldName -> {
                    if (ObjectUtil.isNotEmpty(dict.get(fieldName))) {
                        //put.addColumn(finalCarTripOutput.getTriptype().getBytes(StandardCharsets.UTF_8), fieldName.getBytes(StandardCharsets.UTF_8), dict.get(fieldName).toString().getBytes());
                        put.addColumn(Bytes.toBytes("C1"),fieldName.getBytes(StandardCharsets.UTF_8), dict.get(fieldName).toString().getBytes());
                    }
                });
                table.put(put);
                table.close();
            }
        }

    }
}
